package cqut.wys;

import cqut.wys.config.KafkaClientConfig;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.common.acl.AclBindingFilter;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * Hello world!
 *
 */
public class App 
{

    private static final String NEW_TOPIC ="topic-test2";

    private static final String BROKERURL = "localhost:9092";

    private static AdminClient adminClient = new KafkaClientConfig(BROKERURL).build();

    public static void main(String[] args) {
//        createTopices();
        listTopics();
    }

    //创建topic
    private static void createTopices(){
        //分区数为4，副本因子为1
        NewTopic newTopic = new NewTopic(NEW_TOPIC,4, (short) 1);
        List<NewTopic> newTopicList = new ArrayList<>();
        newTopicList.add(newTopic);
        adminClient.createTopics(newTopicList);
    }

    //删除Topic
    private static void deleteTopices(List<String> deleteTopic){
        adminClient.deleteTopics(deleteTopic);
    }

    //罗列所有Topic
    private static void listTopics(){
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        try {
            Map<String,TopicListing> map = listTopicsResult.namesToListings().get();
            System.out.println(map);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    //查询Topic
    private static void queryTopic(List<String> topicNames){
        adminClient.describeTopics(topicNames);
    }

    //查询集群信息
    private static void describeCluster(){
        adminClient.describeCluster();
    }

    //查询ACL信息
    private static void describeAcls(AclBindingFilter filter){
        adminClient.describeAcls(filter);
    }

    //创建ACL信息
    private static void createAcls(Collection acls){
        adminClient.createAcls(acls);
    }

    //删除ACL信息
    private static void deleteAcls(Collection filters){
        adminClient.deleteAcls(filters);
    }

    //查询配置信息
    private static void describeConfigs(Collection resources){
        adminClient.describeConfigs(resources);
    }

    //修改配置信息
    private static void alterConfigs(Map configs){
        adminClient.alterConfigs(configs);
    }

    //修改副本的日志目录
    private static void alterReplicaLogDirs(Map replicaAssignment){
        adminClient.alterReplicaLogDirs(replicaAssignment);
    }

    //查询节点的日志目录信息
    private static void describeLogDirs(Collection brokers){
        adminClient.describeLogDirs(brokers);
    }

    //查询副本的日志目录信息
    private static void describeReplicaLogDirs(Collection replicas){
        adminClient.describeReplicaLogDirs(replicas);
    }

    //增加分区
    private static void createPartitions(Map newPartitions){
        adminClient.createPartitions(newPartitions);
    }
}
